package test.mq;
import java.io.IOException;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import hyl.base.mq.ISendReceive;
import hyl.base.mq.rabbit.MyRabbitMq;
import hyl.core.MyDate;
import hyl.core.info.CMsg;
import hyl.core.run.IFilter;
import hyl.core.run.IFinishT;
public class testrabbitmq_receive {
	private final static String QUEUE_NAME = "hello";
	public static void main(String[] args) throws java.io.IOException, TimeoutException, KeyManagementException,
			NoSuchAlgorithmException, URISyntaxException {
		// testConsume() ;
		// test一对一();
		 test轮发();
		//test广播();
		// amq.ctag-fNqcIjBRH0J98CPbgHi5pA
	}
	static IFinishT<CMsg> dc = new IFinishT<CMsg>() {
		@Override
		public boolean success(CMsg mp) {
			System.out.println(String.format("1组:%s,%s,%s", mp.s主题, new String( mp.bs内容),
					Thread.currentThread().getName()));
			// MyDate.sleep(100);
			return true;
		}
		@Override
		public void error(CMsg o, Throwable e) {
		}
	};
	static IFinishT<CMsg> dc2 = new IFinishT<CMsg>() {
		@Override
		public boolean success(CMsg mp) {
			System.out.println(String.format("2组:%s,%s,%s", mp.s主题, new String( mp.bs内容),
					Thread.currentThread().getName()));
			// MyDate.sleep(100);
			return true;
		}
		@Override
		public void error(CMsg o, Throwable e) {
		}
	};
	public static void test一对一() {
		try {
			ISendReceive<CMsg> mrm1 = MyRabbitMq.getInstance(MyRabbitMq.M_一对一);
			// 无效的,一对一情况下只会认可一个接收方
			// ISendReceive mrm2 = MyRabbitMq.getInstance(MyRabbitMq.M_一对一);
			mrm1.set主题("语文").set回调函数(dc).receive();
			// mrm2.ready("语文", dc2).receive();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	public static void test轮发() {
		try {
			ISendReceive<CMsg> mrm1 = MyRabbitMq.getInstance(MyRabbitMq.M_轮发);// .set应答方式(false);
			ISendReceive<CMsg> mrm2 = MyRabbitMq.getInstance(MyRabbitMq.M_轮发);// .set应答方式(false);
			// mrm1.receive("jl911", "key911", dc);
			// mrm2.receive("jl911", "key911", dc2);
			mrm1.set主题("分组", "英语").set回调函数(dc,1).receive();
			mrm2.set主题("分组", "英语").set回调函数(dc2,1).receive();
			// MyDate.sleep(5000);
			// mrm2.remove("","key911");
			// mrm2.remove("","key913");
			// mrm2.remove("","hello");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	public static void test广播() {
		try {
			ISendReceive<CMsg> mrm1 = MyRabbitMq.getInstance(MyRabbitMq.M_群发);
			// 无效的,一对一情况下只会认可一个接收方
			ISendReceive<CMsg> mrm2 = MyRabbitMq.getInstance(MyRabbitMq.M_群发);
			mrm1.set主题("分组", "").set回调函数(dc,1).receive();
			mrm2.set主题("分组", "").set回调函数(dc2,1).receive();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	public static void testConsume() {
		/*
		 * ConnectionFactory factory = new ConnectionFactory(); // "guest"/"guest" by
		 * default, limited to localhost connections factory.setUsername("admin");
		 * factory.setPassword("admin"); factory.setVirtualHost("/");
		 * factory.setHost("192.168.1.1"); factory.setPort(5672);
		 */
		ConnectionFactory factory = new ConnectionFactory();
		try {
			factory.setUri("amqp://admin:admin@192.168.0.112:5672");
			// factory.setVirtualHost("/");
			// 打开连接和创建频道，与发送端一样
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();
			// 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；
			// 参数三：是否独占模式；
			// 参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
			channel.queueDeclare(QUEUE_NAME, false, false, false, null);
			// channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
			// String queueName = channel.queueDeclare().getQueue();
			// channel.queueBind(queueName, EXCHANGE_NAME, "");
			// 绑定队列到交换机
			// channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routekey.*");
			//// 同一时刻服务器只会发一条消息给消费者
			// channel.basicQos(1);
			// 创建队列消费者
			System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
			// 指定消费队列
			// 指定消费队列
			// QueueingConsumer consumer = new QueueingConsumer(channel);
			// 监听队列，手动返回完成
			// channel.basicConsume(QUEUE_NAME, false, consumer);
			// QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			// String message = new String(delivery.getBody());
			// DefaultConsumer 异步消费
			channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
						byte[] body) throws IOException {
					String routingKey = envelope.getRoutingKey(); // 队列名称
					String contentType = properties.getContentType(); // 内容类型
					String content = new String(body, "utf-8"); // 消息正文
					System.out.println("消息：" + String.format("routingKey:%s;contentType:%s;consumerTag:%s", routingKey,
							contentType, consumerTag));// UbwCswOG0f521iB7gjYUOg
					System.out.println("消息正文：" + content);
					// 结束接收时调用应答
					// channel.basicAck(envelope.getDeliveryTag(), true);
					// 手动确认消息【参数说明：参数一：该消息的index；
					// 参数二：是否批量应答，true批量确认小于index的消息】
				}
			});
			MyDate.sleep(20000);
			channel.close();
			connection.close();
			System.out.println("wait....");
			// channel.close();
		} catch (KeyManagementException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (NoSuchAlgorithmException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TimeoutException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
